from datetime import datetime
from decimal import ROUND_HALF_UP, Decimal, getcontext
import a_trade_calendar
from fastapi.concurrency import run_in_threadpool
import pandas as pd
from module_stock.dao.monthly_dao import MonthlyDao
from module_stock.dao.weekly_dao import WeeklyDao
from module_stock.entity.vo.monthly_vo import MonthlyModel
from module_stock.entity.vo.weekly_vo import WeeklyModel
from utils.ak_util import fetch_ak_rt_data, fetch_akdata, filter_ak_rt_stock, filter_ak_stock, get_latest_trade_date
from utils.log_util import logger
from numpy import float64
from config.database import AsyncSessionLocal
from config.env import StockConfig
from module_stock.dao.basic_dao import BasicDao
from module_stock.dao.daily_dao import DailyDao
from module_stock.entity.vo.basic_vo import BasicModel
import akshare as ak

from module_stock.entity.vo.daily_vo import DailyModel
from utils.stock_util import CommonUtil

async def sync_basic_job(): 
    """
    从akshare同步股票基础信息
    1.从akshare获取股票列表数据
    2.查询股票基础信息表
    3
    3.1如果没有数据，说明是第一次同步，将akshare的数据直接插入数据库
    3.2如果有数据，则跟akshare的数据进行比对，进行增加、删除操作，说明akshare的增加了说明新股上市
    减少了说明是股票退市
    :param 
    :return
    """
    try:
        async with AsyncSessionLocal() as session:
            datas_ak =  ak.stock_info_a_code_name()
            datas_ak = filter_ak_stock(datas_ak, 'code', 'name')
            symbol_list_ak =  datas_ak['code'].to_list()
            symbol_list_db = await BasicDao.get_all_symbols(session)

            if symbol_list_db and len(symbol_list_db) > 0 :
                #新增股票
                diff_symbols_ak = set(symbol_list_ak).difference(set(symbol_list_db))
                if diff_symbols_ak:
                    diff_datas_ak = datas_ak[datas_ak['code'].isin(diff_symbols_ak)]
                    diff_data_models = []
                    for diff_data in diff_datas_ak.itertuples():   
                        diff_data_models.append(BasicModel(symbol=diff_data.code, name=diff_data.name))
                    await BasicDao.batch_add_basic_dao(session, diff_data_models)
                #退市股票
                diff_symbols_db = set(symbol_list_db).difference(set(symbol_list_ak))
                if diff_symbols_db :
                    await BasicDao.batch_delete_basic_dao(session, diff_symbols_db)
            else:
                #没有数据，直接插入所有数据
                data_models = []
                for data_ak in datas_ak.itertuples():   
                    data_models.append(BasicModel(symbol=data_ak.code, name=data_ak.name))
                await BasicDao.batch_add_basic_dao(session, data_models)
            await session.commit()
    except Exception as e:
        await session.rollback()
        raise e
    
async def sync_daily_job(): 
    """
    从akshare同步股票日线数据, 首先需要依赖基础数据的同步
    1.从akshare获取股票日线数据
    2.查询股票基础信息表
    3
    3.1如果没有数据，说明基础数据表未同步，提示同步基础数据表未同步的错误
    3.2如果有数据，则查询日线表所有的股票去重股票代码，并与基础信息表数据比对
    3.2.1 基础信息表有的，而日线表没有，从akshare进行同步
    3.2.2 基础信息表没有，而日线表有的，从日线表进行删除
    3.2.3 都有的数据，进行迭代获取股票最后一个交易日的akshare股票价格，数据库日线股票价格，
    如果价格不一致，说明股票进行了复权，将该股票在数据库删除，并从akshare获得最新的数据插入数据库
    :param 
    :return: 
    """
    try:
        async with AsyncSessionLocal() as session:
            period='daily'
            symbols_in_basic = await BasicDao.get_all_symbols(session)

            if symbols_in_basic and len(symbols_in_basic) > 0 :
                symbols_in_daily = await DailyDao.get_all_symbols(session)
                
                #新增股票
                diff_symbols_basic = set(symbols_in_basic).difference(set(symbols_in_daily))
                if diff_symbols_basic:                    
                    for diff_symbol_basic in diff_symbols_basic:  
                        data_ak = await run_in_threadpool(fetch_akdata, diff_symbol_basic,StockConfig.sync_ak_startdate,period)
                        data_daily = data_ak.to_dict('records')               
                        await DailyDao.batch_add_daily_dao(session, data_daily)
                    await session.commit()
                #退市股票
                diff_symbols_daily = set(symbols_in_daily).difference(set(symbols_in_basic))
                if diff_symbols_daily :
                    await DailyDao.batch_delete_daily_dao(session, diff_symbols_daily)
                    await session.commit()
                
                            #获得交易时间
                latest_trade_date = get_latest_trade_date()
                
                # 都有的股票
                same_symbols = set(symbols_in_basic).intersection(set(symbols_in_daily))
                if same_symbols :
                    for index, same_symbol in enumerate(same_symbols) :
                        daily_info_db = await DailyDao.get_last_daily_detail(session, same_symbol)
                        if daily_info_db.trade_date == latest_trade_date:
                            continue
                        daily_ak  = await run_in_threadpool(fetch_akdata, same_symbol, daily_info_db.trade_date.strftime('%Y%m%d'), period)
                        daily_ak_sorted = daily_ak.sort_values(by=['trade_date'],ascending=[True])
                        if float64(daily_info_db.close) != daily_ak_sorted.iloc[0]['close'] :
                            await DailyDao.delete_daily_dao(session, DailyModel(symbol=same_symbol))
                            data_ak = await run_in_threadpool(fetch_akdata, same_symbol,StockConfig.sync_ak_startdate,period)
                            data_daily = data_ak.to_dict('records')
                            await DailyDao.batch_add_daily_dao(session, data_daily)
                        elif len(daily_ak_sorted) > 1:                          
                            data_daily = daily_ak_sorted.iloc[1: , :].to_dict('records')
                            await DailyDao.batch_add_daily_dao(session, data_daily) 
                        if index % 10 == 9:
                            logger.info(f'从akshare同步日线数据, 执行到{index+1}次')
                            await session.commit()
            else:
                #没有数据
                raise Exception("股票基础信息表未同步")
            await session.commit()
    except Exception as e:
        await session.rollback()
        logger.exception(e)
        raise e

async def sync_weekly_job(): 
    """
    从akshare同步股票周线数据, 首先需要依赖基础数据的同步
    1.从akshare获取股票周线数据
    2.查询股票基础信息表
    3
    3.1如果没有数据，说明基础数据表未同步，提示同步基础数据表未同步的错误
    3.2如果有数据，则查询周线表所有的股票去重股票代码，并与基础信息表数据比对
    3.2.1 基础信息表有的，而周线表没有，从akshare进行同步
    3.2.2 基础信息表没有，而周线表有的，从周线表进行删除
    3.2.3 都有的数据，进行迭代获取股票最后一个交易日的akshare股票价格，数据库周线股票价格，
    如果价格不一致，说明股票进行了复权，将该股票在数据库删除，并从akshare获得最新的数据插入数据库
    :param 
    :return: 
    """
    try:
        async with AsyncSessionLocal() as session:
            period='weekly'
            symbols_in_basic = await BasicDao.get_all_symbols(session)

            if symbols_in_basic and len(symbols_in_basic) > 0 :
                symbols_in_weekly = await WeeklyDao.get_all_symbols(session)
                
                #新增股票
                diff_symbols_basic = set(symbols_in_basic).difference(set(symbols_in_weekly))
                if diff_symbols_basic:                    
                    for diff_symbol_basic in diff_symbols_basic:  
                        data_ak = await run_in_threadpool(fetch_akdata, diff_symbol_basic,StockConfig.sync_ak_startdate,period)
                        data_weekly = data_ak.to_dict('records')               
                        await WeeklyDao.batch_add_weekly_dao(session, data_weekly)
                    await session.commit()
                #退市股票
                diff_symbols_weekly = set(symbols_in_weekly).difference(set(symbols_in_basic))
                if diff_symbols_weekly :
                    await WeeklyDao.batch_delete_weekly_dao(session, diff_symbols_weekly)
                    await session.commit()
                
                # 都有的股票
                same_symbols = set(symbols_in_basic).intersection(set(symbols_in_weekly))
                if same_symbols :
                    # 先删除当周的数据
                    await WeeklyDao.batch_delete_weekly_dao(session, same_symbols, CommonUtil.get_period_startdate(period))
                    for index, same_symbol in enumerate(same_symbols) :
                        weekly_info_db = await WeeklyDao.get_last_weekly_detail(session, same_symbol)
                        #因为前面统一删除了当周的数据，如果股票是当前周上市的，数据库的数据就是空的
                        if not weekly_info_db:
                            weekly_start_date = CommonUtil.get_period_startdate(period)
                        else :
                            weekly_start_date = weekly_info_db.trade_date.strftime('%Y%m%d')
                        
                        weekly_ak  = await run_in_threadpool(fetch_akdata, same_symbol, weekly_start_date, period)
                        weekly_ak_sorted = weekly_ak.sort_values(by=['trade_date'],ascending=[True])
                        
                        if weekly_info_db:
                            if float64(weekly_info_db.close) != weekly_ak_sorted.iloc[0]['close'] :
                                await WeeklyDao.delete_weekly_dao(session, WeeklyModel(symbol=same_symbol))
                                data_ak = await run_in_threadpool(fetch_akdata, same_symbol,StockConfig.sync_ak_startdate,period)
                                data_weekly = data_ak.to_dict('records')
                                await WeeklyDao.batch_add_weekly_dao(session, data_weekly)
                            elif len(weekly_ak_sorted) > 1:                          
                                data_weekly = weekly_ak_sorted.iloc[1: , :].to_dict('records')
                                await WeeklyDao.batch_add_weekly_dao(session, data_weekly) 
                        else:
                            data_weekly = weekly_ak_sorted.to_dict('records')
                            await WeeklyDao.batch_add_weekly_dao(session, data_weekly) 
                        if index % 10 == 9:
                            logger.info(f'从akshare同步周线数据, 执行到{index+1}次')
                            await session.commit()
            else:
                #没有数据
                raise Exception("股票基础信息表未同步")
            await session.commit()
    except Exception as e:
        await session.rollback()
        logger.exception(e)
        raise e

async def sync_monthly_job(): 
    """
    从akshare同步股票月线数据, 首先需要依赖基础数据的同步
    1.从akshare获取股票月线数据
    2.查询股票基础信息表
    3
    3.1如果没有数据，说明基础数据表未同步，提示同步基础数据表未同步的错误
    3.2如果有数据，则查询月线表所有的股票去重股票代码，并与基础信息表数据比对
    3.2.1 基础信息表有的，而月线表没有，从akshare进行同步
    3.2.2 基础信息表没有，而月线表有的，从月线表进行删除
    3.2.3 都有的数据，进行迭代获取股票最后一个交易日的akshare股票价格，数据库月线股票价格，
    如果价格不一致，说明股票进行了复权，将该股票在数据库删除，并从akshare获得最新的数据插入数据库
    :param 
    :return: 
    """
    try:
        async with AsyncSessionLocal() as session:
            period='monthly'
            symbols_in_basic = await BasicDao.get_all_symbols(session)

            if symbols_in_basic and len(symbols_in_basic) > 0 :
                symbols_in_monthly = await MonthlyDao.get_all_symbols(session)
                
                #新增股票
                diff_symbols_basic = set(symbols_in_basic).difference(set(symbols_in_monthly))
                if diff_symbols_basic:                    
                    for diff_symbol_basic in diff_symbols_basic:  
                        data_ak = await run_in_threadpool(fetch_akdata, diff_symbol_basic,StockConfig.sync_ak_startdate,period)
                        data_monthly = data_ak.to_dict('records')               
                        await MonthlyDao.batch_add_monthly_dao(session, data_monthly)
                    await session.commit()
                #退市股票
                diff_symbols_monthly = set(symbols_in_monthly).difference(set(symbols_in_basic))
                if diff_symbols_monthly :
                    await MonthlyDao.batch_delete_monthly_dao(session, diff_symbols_monthly)
                    await session.commit()
                
                # 都有的股票
                same_symbols = set(symbols_in_basic).intersection(set(symbols_in_monthly))
                if same_symbols :
                    # 先删除当月的数据
                    await MonthlyDao.batch_delete_monthly_dao(session, same_symbols, CommonUtil.get_period_startdate(period))
                    for index, same_symbol in enumerate(same_symbols) :
                        monthly_info_db = await MonthlyDao.get_last_monthly_detail(session, same_symbol)
                        #因为前面统一删除了当月的数据，如果股票是当前月上市的，数据库的数据就是空的
                        if not monthly_info_db:
                            monthly_start_date = CommonUtil.get_period_startdate(period)
                        else :
                            monthly_start_date = monthly_info_db.trade_date.strftime('%Y%m%d')
                        monthly_ak  = await run_in_threadpool(fetch_akdata, same_symbol, monthly_start_date, period)
                        monthly_ak_sorted = monthly_ak.sort_values(by=['trade_date'],ascending=[True])
                        if monthly_info_db:
                            if float64(monthly_info_db.close) != monthly_ak_sorted.iloc[0]['close'] :
                                await MonthlyDao.delete_monthly_dao(session, MonthlyModel(symbol=same_symbol))
                                data_ak = await run_in_threadpool(fetch_akdata, same_symbol,StockConfig.sync_ak_startdate,period)
                                data_monthly = data_ak.to_dict('records')
                                await MonthlyDao.batch_add_monthly_dao(session, data_monthly)
                            elif len(monthly_ak_sorted) > 1:                          
                                data_monthly = monthly_ak_sorted.iloc[1: , :].to_dict('records')
                                await MonthlyDao.batch_add_monthly_dao(session, data_monthly) 
                        else :
                            data_monthly = monthly_ak_sorted.to_dict('records')
                            await MonthlyDao.batch_add_monthly_dao(session, data_monthly) 
                        if index % 10 == 9:
                            logger.info(f'从akshare同步月线数据, 执行到{index+1}次')
                            await session.commit()
            else:
                #没有数据
                raise Exception("股票基础信息表未同步")
            await session.commit()
    except Exception as e:
        await session.rollback()
        logger.exception(e)
        raise e

async def sync_daily_from_realtime_job(): 
    """
    从akshare的实时行情中同步股票最新日线数据, 首先需要依赖基础数据的同步
    1.从akshare获取股票实时日线数据
    1.1过滤st等股票
    1.2从最新价、成交额、成交量过滤停复牌股票
    1.3从akshare查询停复牌信息，过滤停复牌股票
    1.4从akshare查询分红派息除权信息，过滤除权股票
    2.迭代除权股票，删除数据库日线表除权股票数据，从akshare查询所有历史日线行情数据，并插入日线数据库

    3.查询股票基础信息表
    3.1如果没有数据，说明基础数据表未同步，提示同步基础数据表未同步的错误
    3.2如果有数据，则查询日线表所有的股票去重股票代码，并与基础信息表数据比对
    3.2.1 基础信息表有的，而日线表没有，从实时日线数据获取信息并插入数据库
    3.2.2 基础信息表没有，而日线表有的，从日线表进行删除
    3.2.3 都有的数据，进行迭代获取股票最后一个交易日的akshare股票价格，数据库日线股票价格，
    如果价格不一致，说明股票进行了复权，将该股票在数据库删除，并从akshare获得最新的数据插入数据库
    :param 
    :return: 
    """
    try:
        async with AsyncSessionLocal() as session:
            getcontext().prec = 20  # 足够覆盖2位小数计算
            getcontext().rounding = ROUND_HALF_UP  # 银行家舍入

            #获得交易时间
            latest_trade_date = get_latest_trade_date()
                        
            period='daily'
            symbols_in_basic = await BasicDao.get_all_symbols(session)

            if symbols_in_basic and len(symbols_in_basic) > 0 :
                symbols_in_daily = await DailyDao.get_all_symbols(session)
                
                #新增股票
                diff_symbols_basic = set(symbols_in_basic).difference(set(symbols_in_daily))
                if diff_symbols_basic:                    
                    for diff_symbol_basic in diff_symbols_basic:  
                        data_ak = await run_in_threadpool(fetch_akdata, diff_symbol_basic,StockConfig.sync_ak_startdate,period)
                        data_daily = data_ak.to_dict('records')               
                        await DailyDao.batch_add_daily_dao(session, data_daily)
                    await session.commit()
                #退市股票
                diff_symbols_daily = set(symbols_in_daily).difference(set(symbols_in_basic))
                if diff_symbols_daily :
                    await DailyDao.batch_delete_daily_dao(session, diff_symbols_daily)
                    await session.commit()
                
                # 都有的股票
                same_symbols = set(symbols_in_basic).intersection(set(symbols_in_daily))
                if same_symbols :
                    #实时行情
                    rt_stock_df = await run_in_threadpool(fetch_ak_rt_data)    
                    # rt_stock_df = pd.DataFrame([{
                    #     "序号" : "32973732",
                    #     "symbol" : "600859",
                    #     "name" : "王府井",
                    #     "close" : Decimal(14.28).quantize(Decimal('0.00')),
                    #     "pct_chg" : Decimal(-0.90).quantize(Decimal('0.00')),
                    #     "chg" : Decimal(-0.13).quantize(Decimal('0.00')),
                    #     "vol" : Decimal(174888.00).quantize(Decimal('0.00')),
                    #     "amount" : Decimal(250268285.00).quantize(Decimal('0.00')),
                    #     "amplitude" : Decimal(1.53).quantize(Decimal('0.00')),
                    #     "high" : Decimal(14.45).quantize(Decimal('0.00')),
                    #     "low" : Decimal(14.23).quantize(Decimal('0.00')),
                    #     "open": Decimal(14.38).quantize(Decimal('0.00')),
                    #     "pre_close": Decimal(14.41).quantize(Decimal('0.00')),
                    #     "turn": Decimal(1.56).quantize(Decimal('0.00'))
                    # }]
                    # )    
                    rt_stock_df = filter_ak_stock(rt_stock_df, 'symbol', 'name')
                    rt_stock_df = filter_ak_rt_stock(rt_stock_df)
                    selected_columns = ['trade_date', 'symbol', 'open','close',
                            'low', 'high', 'vol', 'amount', 'amplitude', 'pct_chg', 'chg', 'turn'] 
                    transform_columns = ['open', 'close', 'pre_close', 'high', 
                                        'low', 'vol', 'amount', 'amplitude', 'pct_chg', 'chg', 'turn']
                    rt_stock_df[transform_columns] = rt_stock_df[transform_columns].map(
                        lambda x: (Decimal(str(x)) * Decimal('1.00')).quantize(Decimal('0.00'), rounding=ROUND_HALF_UP)
                    )
                    rt_stock_df['trade_date'] = latest_trade_date

                    for index, same_symbol in enumerate(same_symbols) :
                        daily_info_db = await DailyDao.get_last_daily_detail(session, same_symbol)
                        if (daily_info_db.trade_date == latest_trade_date) :
                            continue
                        daily_ak  = rt_stock_df[rt_stock_df['symbol'] == same_symbol]
                        if daily_ak.empty:
                            continue
                        if daily_info_db.close != daily_ak.iloc[0]['pre_close'] :
                            await DailyDao.delete_daily_dao(session, DailyModel(symbol=same_symbol))
                            data_ak = await run_in_threadpool(fetch_akdata, same_symbol,StockConfig.sync_ak_startdate,period)
                            data_daily = data_ak.to_dict('records')
                            await DailyDao.batch_add_daily_dao(session, data_daily)
                        else :
                            data_daily = daily_ak[selected_columns].to_dict('records')
                            await DailyDao.batch_add_daily_dao(session, data_daily) 
                        if index % 10 == 9:
                            logger.info(f'从akshare同步日线数据, 执行到{index+1}次')
                            await session.commit()
            else:
                #没有数据
                raise Exception("股票基础信息表未同步")
            await session.commit()
    except Exception as e:
        await session.rollback()
        logger.exception(e)
        raise e
